package com.example.test5

import akka.actor.{Actor, PoisonPill}

case class Register(key: String)
case class Beat(key: String)
case object Ack

class Master extends Actor {
  // 超时时间阈值
  val DURATION_THRESHOLD = 5 // seconds

  // 存放注册的 workers
  val workers = scala.collection.mutable.HashMap[String, Long]()

  // 启动心跳定时检查进程
  val thread = new Thread {
    override def run(): Unit = {
      println("start master check loop....")

      while (true) {
        Thread.sleep(1000)
        checkTimestamp()
      }
    }
  }
  thread.start

  override def receive: Receive = {
    case Register(key) =>
      if (workers.contains(key)) {
        println(s"worker[$key] register conflict!")
      } else {
        updateTimestamp(key)
        println(s"worker[$key] register successful.")
        sender() ! Ack
      }
    case Beat(key) =>
      if (workers.contains(key)) { // 响应 worker 心跳，更新时间戳
        updateTimestamp(key)
        sender() ! Ack
      } else { // 关闭超时的worker
        sender() ! PoisonPill
      }
  }

  /**
   * 更新心跳时间
   * @param key
   */
  private def updateTimestamp(key: String) = {
    this.synchronized {
      workers(key) = System.currentTimeMillis()
    }
  }

  /**
   * 定时检查心跳时间，将超时的从 hashmap 中删除
   */
  private def checkTimestamp(): Unit = {
    this.synchronized {
      val now = System.currentTimeMillis()

      println("检查心跳时间戳：")
      for ((worker, ts) <- workers.toSet) {
        val duration = (now - ts) / 1000
        println(s"$worker => $duration 秒")
        if (duration > DURATION_THRESHOLD) {
          println(s"$worker 已经超时")
          workers.remove(worker)
        }
      }
    }
  }
}